package jobs.job3.classloaderdemo.client;

import jobs.job3.classloaderdemo.server.standard.bean.TaskBean;
import jobs.job3.classloaderdemo.server.zkutil.ZkClient;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

/*
用于完成 任务 和 计算节点 之间的一个分配关系

O 表示普通任务的输出
M 表示reduceTask的输入，是通过所有普通任务的输出O决定的

 */
public class Client<M,O> {
//    通过zk获取动态的计算节点列表
    ZkClient zkClient = new ZkClient();
    public Client() {
        zkClient.connectZk();
    }

    /**
     * @param subTasks  普通任务列表，该例中为2个排序任务
     * @param converter 用于将多个O转换成reduceTask的输入
     * @param reduceTask 聚合任务
     * @return          reduceTask(t1(a),t1(b))
     */
    public O execute(List<TaskBean> subTasks,
                     Function<List<O>, M> converter,
                     TaskBean reduceTask) throws Exception {
        List<CNClient<O>> cnClients = new ArrayList<>();
//        根据端口号绑定到一个计算节点上
//        for (String cnHost : cnHosts) {
//            String[] hostPort = cnHost.split(":");
//            CNClient<O> cnClient = new CNClient<>(hostPort[0], Integer.parseInt(hostPort[1]));
//            cnClients.add(cnClient);
//        }
        List<String> serverInfos = zkClient.getServerInfos();
        for (String info : serverInfos) {
            String[] hostPort = info.split(":");
            CNClient<O> cnClient = new CNClient<>(hostPort[0], Integer.parseInt(hostPort[1]));
            cnClients.add(cnClient);
        }

        CountDownLatch cdl = new CountDownLatch(subTasks.size());
//        收集所有任务的返回结果
        List<O> oList = new LinkedList<>();
        final int[] index = {0};
//        通过轮询的方法给计算节点分配任务
        for (TaskBean subTask : subTasks) {
            new Thread(() -> {
                try {
                    O o = cnClients.get(index[0]++ % cnClients.size())
                            .submitAndExecute(subTask);
                    oList.add(o);
                    cdl.countDown();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }).start();
        }
        cdl.await();//阻塞直到所有的任务被执行完成
        M m = converter.apply(oList);//将所有任务的计算结果转换成reduceTask的输入
        reduceTask.setInput(m);
        return cnClients.get(index[0] % cnClients.size())
                .submitAndExecute(reduceTask); //继续使用轮询的方式让下一个计算节点执行 reduceTask
    }

}





